package zhao.exmaple.watearkDemo;

import cn.hutool.core.util.StrUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class SocketDemo {
//    jenkins虚拟机服务
//[root@jenkins ~]# nc -lk 7777
    public static void main(String[] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Configuration configuration = new Configuration();
//        http://localhost:8081/#/job/81614f6dfc8710d525bc87e850b5459f/overview
        configuration.setInteger("rest.port", 8081);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
        env.setParallelism(1);

        SingleOutputStreamOperator<Integer> os = env.socketTextStream("jenkins", 7777)
                .map(new RichMapFunction<String, Integer>() {
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        int indexOfThisSubtask = getRuntimeContext().getIndexOfThisSubtask();
                        String format = StrUtil.format("open 子任务的编号{} 子任务的名称{}", indexOfThisSubtask, getRuntimeContext().getTaskNameWithSubtasks());
                        System.out.println(format);
                    }

                    // 程序正常退出比如数据服务关闭等，异常退出不会执行
                    @Override
                    public void close() throws Exception {
                        int indexOfThisSubtask = getRuntimeContext().getIndexOfThisSubtask();
                        String format = StrUtil.format("close 子任务的编号{} 子任务的名称{}", indexOfThisSubtask, getRuntimeContext().getTaskNameWithSubtasks());
                        System.out.println(format);
                    }

                    @Override
                    public Integer map(String value) throws Exception {
                        return Integer.valueOf(value) +  1;
                    }
                });

        os.print();
        env.execute();
    }
}
